"""
Case Type   : 发布订阅冲突解决
Case Name   : 违反订阅端非空约束，需要手动解决冲突
Create At   : 2024/04/15
Owner       : @songjing20
Description :
    1.订阅端设置subscription_conflict_resolution=keep_local
    2.集群1创建表和发布
    3.集群2创建表(非空约束)和订阅
    4.发布端表中插入数据
    5.订阅端查询表数据是否同步，插入数据
    6.发布端插入数据使订阅端违反非空约束
    7.订阅端查询数据是否同步
    8.订阅跳过空值事务commit_lsn，再次查询数据
    9.发布端更新数据与订阅冲突
    10.订阅查询表数据，跳过冲突事务commit_lsn，再次查询
    11.清理环境
Expect      :
    1.成功
    2.成功
    3.成功
    4.成功
    5.同步,(1,pub1)，插入数据成功
    6.成功
    7.不同步
    8.成功，(3,ok)同步到订阅端
    9.成功
    10.不同步，跳过成功，数据同步
    11.成功
History     :
"""

import os
import time
import unittest

from yat.test import macro
from yat.test import Node
from testcase.utils.Logger import Logger
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Common import Common
from testcase.utils.Constant import Constant

Primary_SH = CommonSH('PrimaryDbUser')


@unittest.skipIf(3 > Primary_SH.get_node_num(), '非1+2以上环境不执行')
class Pubsubclass(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f"-----{os.path.basename(__file__)} start-----")
        self.pri_userdb_pub = Node(node='PrimaryDbUser')
        self.pri_userdb_sub = Node(node='remote1_PrimaryDbUser')
        self.constant = Constant()
        self.commsh_pub = CommonSH('PrimaryDbUser')
        self.commsh_sub = CommonSH('remote1_PrimaryDbUser')
        self.com_pub = Common()
        self.com_sub = Common('remote1_PrimaryDbUser')
        self.tb_name = 't_pubsub_0357'
        self.pubname = 'pub_pubsub_0357'
        self.subname = 'sub_pubsub_0357'
        self.conflict_param = 'subscription_conflict_resolution'
        self.para_value = self.com_sub.show_param(
            self.conflict_param, macro.DB_ENV_PATH_REMOTE1)
        self.port = str(int(self.pri_userdb_pub.db_port) + 1)
        self.wal_level = self.com_pub.show_param("wal_level")
        self.user_param_pub = f'-U {self.pri_userdb_pub.db_user} ' \
                              f'-W {self.pri_userdb_pub.db_password}'
        self.user_param_sub = f'-U {self.pri_userdb_sub.db_user} ' \
                              f'-W {self.pri_userdb_sub.db_password}'
        text = '-----查询集群状态，修改pg_hba，发布端设置wal_level=logical-----'
        self.log.info(text)
        self.log.info('----- 发布端状态 -----')
        status = self.commsh_pub.get_db_cluster_status('detail')
        self.log.info(status)
        self.assertEqual(status.count('Normal'), 4, f'执行失败:{text}')
        self.log.info('----- 订阅端状态 -----')
        status = self.commsh_sub.get_db_cluster_status(
            'detail', env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(status)
        self.assertEqual(status.count('Normal'), 4, f'执行失败:{text}')
        guc_res = self.commsh_pub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host  replication  {self.pri_userdb_sub.db_user}  '
            f'{self.pri_userdb_sub.db_host}/32 sha256')
        self.log.info(guc_res)
        self.assertTrue(guc_res, f'发布端配置流复制权限失败:{text}')
        if 'logical' != self.wal_level:
            res = self.commsh_pub.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG, 'wal_level=logical')
            self.assertTrue(res, f'发布端设置wal_level失败:{text}')
            restart_res = self.commsh_pub.restart_db_cluster()
            self.assertTrue(restart_res, f'重启集群失败:{text}')
        guc_res = self.commsh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE1,
            f'host  replication  {self.pri_userdb_pub.db_user}  '
            f'{self.pri_userdb_pub.db_host}/32 sha256',
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(guc_res)
        self.assertTrue(guc_res, f'订阅端配置流复制权限失败:{text}')

    def test_pubsub(self):
        text = '-----step1:订阅设置subscription_conflict_resolution=keep_local;' \
               'expect:成功-----'
        self.log.info(text)
        res = self.commsh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG,
            f'{self.conflict_param}=keep_local',
            dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(res)
        self.assertTrue(res, f'执行失败:{text}')

        text = '-----step2:集群1创建表和发布;expect:成功-----'
        self.log.info(text)
        sql = f"drop table if exists {self.tb_name} cascade;" \
              f"create table {self.tb_name}(" \
              f"c_int int primary key, c_varchar varchar(15));" \
              f"drop publication if exists {self.pubname};" \
              f"create publication {self.pubname} for table {self.tb_name};"
        pub_res = self.commsh_pub.execut_db_sql(sql, self.user_param_pub)
        self.log.info(pub_res)
        self.assertEqual(pub_res.count(self.constant.CREATE_TABLE_SUCCESS), 2,
                         f'发布端建表失败:{text}')
        self.assertIn(self.constant.create_pub_succ_msg, pub_res,
                      f'执行失败:{text}')

        text = '-----step3:集群2创建表(非空约束)和订阅;expect:成功-----'
        self.log.info(text)
        guc_res = self.commsh_sub.execute_generate(
            macro.COMMON_PASSWD, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertIn(self.constant.create_keycipher_success, guc_res,
                      f'执行失败:{text}')
        sql = f"drop table if exists {self.tb_name} cascade;" \
              f"create table {self.tb_name}(c_int int primary key, " \
              f"c_varchar varchar(15) not null default 'sub');" \
              f"drop subscription if exists {self.subname} cascade;" \
              f"create subscription {self.subname} connection " \
              f"'host={self.pri_userdb_pub.db_host} " \
              f"port={self.port} " \
              f"user={self.pri_userdb_pub.db_user} " \
              f"dbname={self.pri_userdb_pub.db_name} " \
              f"password={self.pri_userdb_pub.ssh_password}' " \
              f"publication {self.pubname};"
        sub_res = self.commsh_sub.execut_db_sql(
            sql, self.user_param_sub, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_res)
        self.assertEqual(pub_res.count(self.constant.CREATE_TABLE_SUCCESS), 2,
                         f'订阅端建表失败:{text}')
        self.assertIn(self.constant.create_sub_succ_msg, sub_res,
                      f'创建订阅失败:{text}')

        text = '-----step4:发布端表中插入数据;expect:成功-----'
        self.log.info(text)
        insert_res = self.commsh_pub.execut_db_sql(
            f"insert into {self.tb_name} values(1, 'pub1');",
            self.user_param_pub)
        self.log.info(insert_res)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG, insert_res,
                      f'执行失败:{text}')

        text = '-----step5:订阅端查询是否同步，表中插入数据;expect:同步，成功-----'
        self.log.info(text)
        self.commsh_pub.check_sub_consistency(
            self.subname, self.user_param_pub, 1200)
        time.sleep(5)
        sql = f"select * from {self.tb_name};" \
              f"insert into {self.tb_name} values(2, 'not null');"
        sql_res = self.commsh_sub.execut_db_sql(
            sql, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sql_res)
        self.assertIn('1 | pub1\n(1 row)', sql_res, f'执行失败:{text}')
        self.assertIn(self.constant.INSERT_SUCCESS_MSG, sql_res,
                      f'执行失败:{text}')

        text = '-----step6:发布端插入数据使订阅端违反非空约束;expect:成功-----'
        self.log.info(text)
        pub_res = self.commsh_pub.execut_db_sql(
            f"insert into {self.tb_name} values(2, '');"
            f"insert into {self.tb_name} values(3, 'ok');"
            f"select gs_xlogdump_xid(xmin) from {self.tb_name} where c_int=2;",
            self.user_param_pub)
        self.log.info(pub_res)
        file = pub_res.splitlines()[-2].strip()
        self.log.info(file)
        cmd = f"grep '^start_lsn' {file} | sed -n '3p' | awk '{{{{print $2}}}}'"
        self.log.info(cmd)
        lsn = self.pri_userdb_pub.sh(cmd).result().strip()
        self.assertEqual(pub_res.count(self.constant.INSERT_SUCCESS_MSG), 2,
                         f'执行失败:{text}')

        text = '-----step7:订阅端查询表数据是否同步;expect:不同步-----'
        self.log.info(text)
        self.commsh_pub.check_sub_consistency(
            self.subname, self.user_param_pub, 20)
        select_sql = f"select * from {self.tb_name} order by 1;"
        select_res = self.commsh_sub.execut_db_sql(
            select_sql, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(select_res)
        self.assertIn('1 | pub1\n     2 | not null\n(2 rows)', select_res,
                      f'执行失败:{text}')

        text = '-----step8:订阅跳过空值事务commit_lsn，再次查询数据;' \
               'expect:成功，(3,ok)同步到订阅端-----'
        self.log.info(text)
        sub_res = self.commsh_sub.execut_db_sql(
            f"alter subscription {self.subname} set(skiplsn = '{lsn}');"
            f"select pg_sleep(10);"
            f"select * from {self.tb_name} order by 1;",
            self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_res)
        self.assertIn(self.constant.alter_sub_succ_msg, res, f'执行失败:{text}')
        msg = '1 | pub1\n     2 | not null\n     3 | ok\n(3 rows)'
        self.assertIn(msg, res, f'执行失败:{text}')

        text = '-----step9:发布端更新数据与订阅冲突;expect:3条数据-----'
        pub_res = self.commsh_pub.execut_db_sql(
            f"update {self.tb_name} set c_varchar='' where c_int = 1;"
            f"insert into {self.tb_name} values(123, 'tbpub');"
            f"select gs_xlogdump_xid(xmin) from {self.tb_name} where c_int=1;",
            self.user_param_pub)
        self.log.info(pub_res)
        file = pub_res.splitlines()[-2].strip()
        self.log.info(file)
        lsn = self.pri_userdb_pub.sh(cmd).result().strip()
        self.assertIn(self.constant.UPDATE_SUCCESS_MSG, pub_res,
                      f'执行失败:{text}')
        self.assertIn(self.constant.INSERT_SUCCESS_MSG, pub_res,
                      f"执行失败:{text}")

        text = '-----step10:订阅查询表数据，跳过冲突事务commit_lsn，再次查询;' \
               'expect:不同步，跳过成功，数据同步-----'
        self.log.info(text)
        select_res = self.commsh_sub.execut_db_sql(
            select_sql, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(select_res)
        self.assertIn(msg, select_res, f'执行失败:{text}')
        sub_res = self.commsh_sub.execut_db_sql(
            f"alter subscription {self.subname} set(skiplsn = '{lsn}');"
            f"select pg_sleep(5);"
            f"select * from {self.tb_name} order by 1;",
            self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_res)
        self.assertIn(self.constant.alter_sub_succ_msg, sub_res, f'执行失败:{text}')
        self.assertIn('   '.join([msg[:-8], '123 | tbpub\n(4 rows)']), sub_res,
                      f'执行失败:{text}')

    def tearDown(self):
        text = '-----step11:清理环境;expect:成功-----'
        self.log.info(text)
        drop_sub_res = self.commsh_sub.execut_db_sql(
            f"drop subscription if exists {self.subname};",
            self.user_param_sub, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(drop_sub_res)
        drop_pub_res = self.commsh_pub.execut_db_sql(
            f"drop publication if exists {self.pubname};", self.user_param_pub)
        self.log.info(drop_pub_res)
        drop_tb = f"drop table if exists {self.tb_name} cascade;"
        drop_tb_sub = self.commsh_sub.execut_db_sql(
            drop_tb, self.user_param_sub, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(drop_tb_sub)
        drop_tb_pub = self.commsh_pub.execut_db_sql(
            drop_tb, self.user_param_pub)
        self.log.info(drop_tb_pub)
        wal_level_res = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'wal_level={self.wal_level}')
        self.log.info(wal_level_res)
        restart_res = self.commsh_pub.restart_db_cluster()
        pub_guc_res = self.commsh_pub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host    replication  {self.pri_userdb_sub.db_user} '
            f'{self.pri_userdb_sub.db_host}/32 ')
        self.log.info(pub_guc_res)
        sub_guc_res = self.commsh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE1,
            f'host    replication  {self.pri_userdb_pub.db_user} '
            f'{self.pri_userdb_pub.db_host}/32',
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_guc_res)
        res = self.commsh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG,
            f'{self.conflict_param} = {self.para_value}',
            dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(res)
        self.assertTrue(res, f'执行失败:{text}')
        self.assertIn(self.constant.drop_pub_succ_msg, drop_pub_res,
                      f'删除发布失败:{text}')
        self.assertIn(self.constant.drop_sub_succ_msg, drop_sub_res,
                      f'删除订阅失败:{text}')
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, drop_tb_pub,
                      f'发布端删表失败:{text}')
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, drop_tb_sub,
                      f'订阅端删表失败:{text}')
        self.assertTrue(wal_level_res, f'执行失败:{text}')
        self.assertTrue(restart_res, f'重启发布端集群失败:{text}')
        self.assertTrue(pub_guc_res, f'执行失败:{text}')
        self.assertTrue(sub_guc_res, f'执行失败:{text}')
        self.log.info(f"-----{os.path.basename(__file__)} end-----")
